package com.gomro.galaxy.foundation.bi.engine.analysis;

import cn.gomro.cloud.core.common.utils.DateUtils;
import cn.gomro.cloud.core.common.utils.EncryptUtils;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.gomro.galaxy.core.batch.ElasticSearchClzItemReader;
import com.gomro.galaxy.core.batch.ElasticSearchClzItemWriter;
import com.gomro.galaxy.core.commons.elastic.ElasticSearchRest;
import com.gomro.galaxy.foundation.bi.domain.Constants;
import com.gomro.galaxy.foundation.bi.domain.event.VendorStatisticsEvent;
import com.gomro.galaxy.foundation.bi.domain.model.*;
import com.gomro.galaxy.foundation.bi.domain.model.analysis.VendorStatistics;
import com.gomro.galaxy.foundation.bi.engine.AnalysisLastTime;
import com.gomro.galaxy.foundation.bi.engine.model.AnalysisCache;
import com.gomro.galaxy.foundation.bi.engine.query.*;
import com.gomro.galaxy.foundation.bi.engine.util.BoolQueryBuilders;
import com.gomro.galaxy.foundation.bi.engine.util.EngineUtils;
import com.gomro.galaxy.foundation.bi.exception.BiException;
import lombok.extern.slf4j.Slf4j;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.action.search.ShardSearchFailure;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.common.unit.TimeValue;
import org.elasticsearch.index.query.BoolQueryBuilder;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.aggregations.Aggregation;
import org.elasticsearch.search.aggregations.AggregationBuilder;
import org.elasticsearch.search.aggregations.AggregationBuilders;
import org.elasticsearch.search.aggregations.bucket.filter.ParsedFilter;
import org.elasticsearch.search.aggregations.metrics.Cardinality;
import org.elasticsearch.search.aggregations.metrics.Sum;
import org.elasticsearch.search.aggregations.metrics.ValueCount;
import org.elasticsearch.search.sort.SortBuilders;
import org.elasticsearch.search.sort.SortOrder;
import org.springframework.batch.core.*;
import org.springframework.batch.core.configuration.annotation.JobBuilderFactory;
import org.springframework.batch.core.configuration.annotation.StepBuilderFactory;
import org.springframework.batch.core.configuration.annotation.StepScope;
import org.springframework.batch.core.launch.support.RunIdIncrementer;
import org.springframework.batch.item.ItemProcessor;
import org.springframework.batch.item.ItemReader;
import org.springframework.batch.item.ItemWriter;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.cloud.bus.BusProperties;
import org.springframework.context.ApplicationEventPublisher;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.lang.NonNull;
import org.springframework.util.CollectionUtils;

import java.time.LocalDateTime;
import java.time.ZoneId;
import java.util.*;
import java.util.stream.Collectors;

import static com.gomro.galaxy.foundation.bi.engine.util.EngineUtils.MAX_BATCH_NUM;

/**
 * 供应商数据统计job
 */
@Slf4j
@Configuration
public class VendorStatisticsJob {
    private static final String BiAnalysisLastName = Constants.BiDataMartIndexPrefix + AnalysisLastTime.TableName;
    private static final String BiWarehouseOrganizationName = Constants.BiWarehouseIndexPrefix + AcquireOrganization.TableName;
    private static final String BiWarehouseQuotationGoodsName = Constants.BiWarehouseIndexPrefix + AcquireQuotationGoods.TableName;
    private static final String BiWarehouseVendorOrderName = Constants.BiWarehouseIndexPrefix + AcquireVendorOrder.TableName;
    private static final String BiWarehouseGoodsName = Constants.BiWarehouseIndexPrefix + AcquireGoods.TableName;
    private static final String BiWarehouseGoodsProfileName = Constants.BiWarehouseIndexPrefix + AcquireGoodsProfile.TableName;
    private static final String BiWarehousePurchaseGoodsName = Constants.BiWarehouseIndexPrefix + AcquirePurchaseGoods.TableName;
    private static final String BiWarehousePlatformGrantUserName = Constants.BiWarehouseIndexPrefix + AcquirePlatformGrantUser.TableName;
    private static final String BiWarehouseMemberName = Constants.BiWarehouseIndexPrefix + AcquireMember.TableName;
    private static final String BiDataMartName = Constants.BiDataMartIndexPrefix + VendorStatistics.TableName;
    private static final String JobName = Constants.JobBeanNamePrefix + VendorStatistics.TableName;
    private static final String StepName1 = Constants.StepBeanNamePrefix + VendorStatistics.TableName + 1;
    private static final String StepName2 = Constants.StepBeanNamePrefix + VendorStatistics.TableName + 2;
    private static final String StepName3 = Constants.StepBeanNamePrefix + VendorStatistics.TableName + 3;
    private static final String StepName4 = Constants.StepBeanNamePrefix + VendorStatistics.TableName + 4;
    private static final String StepName5 = Constants.StepBeanNamePrefix + VendorStatistics.TableName + 5;
    private static final String StepName6 = Constants.StepBeanNamePrefix + VendorStatistics.TableName + 6;
    private static final String StepName7 = Constants.StepBeanNamePrefix + VendorStatistics.TableName + 7;

    private static final String ReaderName1 = Constants.ReaderBeanNamePrefix + VendorStatistics.TableName + 1;
    private static final String ReaderName2 = Constants.ReaderBeanNamePrefix + VendorStatistics.TableName + 2;
    private static final String ReaderName3 = Constants.ReaderBeanNamePrefix + VendorStatistics.TableName + 3;
    private static final String ReaderName4 = Constants.ReaderBeanNamePrefix + VendorStatistics.TableName + 4;
    private static final String ReaderName5 = Constants.ReaderBeanNamePrefix + VendorStatistics.TableName + 5;
    private static final String WriterName1 = Constants.WriterBeanNamePrefix + VendorStatistics.TableName + 1;
    private static final String WriterName2 = Constants.WriterBeanNamePrefix + VendorStatistics.TableName + 2;
    private static final String WriterName3 = Constants.WriterBeanNamePrefix + VendorStatistics.TableName + 3;
    private static final String WriterName4 = Constants.WriterBeanNamePrefix + VendorStatistics.TableName + 4;
    private static final String WriterName5 = Constants.WriterBeanNamePrefix + VendorStatistics.TableName + 5;
    private static final String ProcessName1 = Constants.ProcessBeanNamePrefix + VendorStatistics.TableName + 1;
    private static final String ProcessName2 = Constants.ProcessBeanNamePrefix + VendorStatistics.TableName + 2;
    private static final String ProcessName3 = Constants.ProcessBeanNamePrefix + VendorStatistics.TableName + 3;
    private static final String ProcessName4 = Constants.ProcessBeanNamePrefix + VendorStatistics.TableName + 4;
    private static final String ProcessName5 = Constants.ProcessBeanNamePrefix + VendorStatistics.TableName + 5;
    private static final String ProcessName6 = Constants.ProcessBeanNamePrefix + VendorStatistics.TableName + 6;
    private static final String ProcessName7 = Constants.ProcessBeanNamePrefix + VendorStatistics.TableName + 7;
    private static final String PARTICIPATE_NUM = "participateNum";
    private static final String QUOTATION_NUM = "quotationNum";
    private static final String PICKETED_NUM = "picketedNum";
    private static final String UNIT_TIME_QUOTATION_NUM = "unitTimeQuotationNum";
    private static final String ID_NUM = "idNum";

    private ApplicationEventPublisher applicationEventPublisher;
    private BusProperties busProperties;
    private JobBuilderFactory jobBuilderFactory;
    private StepBuilderFactory stepBuilderFactory;
    private ElasticSearchRest rest;
    private RestHighLevelClient client;
    private ObjectMapper mapper;
    private AnalysisCache<VendorStatistics> cache = new AnalysisCache<>();

    public VendorStatisticsJob(ApplicationEventPublisher applicationEventPublisher, BusProperties busProperties, JobBuilderFactory jobBuilderFactory, StepBuilderFactory stepBuilderFactory, RestHighLevelClient client, ObjectMapper mapper) {
        this.applicationEventPublisher = applicationEventPublisher;
        this.busProperties = busProperties;
        this.jobBuilderFactory = jobBuilderFactory;
        this.stepBuilderFactory = stepBuilderFactory;
        this.client = client;
        this.mapper = mapper;
        this.rest = new ElasticSearchRest(client, mapper);
    }

    private JobExecutionListener jobExecutionListener = new JobExecutionListener() {

        @Override
        public void beforeJob(@NonNull JobExecution jobExecution) {

            ElasticSearchRest rest = new ElasticSearchRest(client, mapper);

            // 初始化数据索引
            if (!rest.indexExists(BiDataMartName)) {

                rest.indexCreate(BiDataMartName);
                rest.indexMappingPut(BiDataMartName, "analysis/" + BiDataMartName + ".json", true);
                rest.indexFlush(BiDataMartName, null, null);
            }

            // 初始化最后更新时间索引
            if (!rest.indexExists(BiAnalysisLastName)) {

                rest.indexCreate(BiAnalysisLastName);
                rest.indexMappingPut(BiAnalysisLastName, "analysis/" + BiAnalysisLastName + ".json", true);
                rest.indexFlush(BiAnalysisLastName, null, null);
            }
        }

        @Override
        public void afterJob(@NonNull JobExecution jobExecution) {
            if (Objects.nonNull(cache.getCache())) {
                log.debug("cache size -> {}", cache.getCache().size());
                cache.getCache().clear();
            }
            log.debug("last time ---> {}", DateUtils.dateToStr(cache.getLast()));
            // 回写最后更新时间
            if (jobExecution.getStatus() == BatchStatus.COMPLETED) {
                rest.documentCreate(BiAnalysisLastName, VendorStatistics.TableName, AnalysisLastTime.builder().id(VendorStatistics.TableName).last(cache.getLast()).build());
                applicationEventPublisher.publishEvent(new VendorStatisticsEvent(cache.getLast()));
            }
        }
    };

    @Bean(ReaderName1)
    @StepScope
    public ElasticSearchClzItemReader<AcquireOrganization> reader1() {

        // 获取最后更新时间
        AnalysisLastTime analysisLastTime = rest.documentGet(BiAnalysisLastName, VendorStatistics.TableName, AnalysisLastTime.class);
        Date start;
        if (analysisLastTime != null && analysisLastTime.getLast() != null) {
            start = DateUtils.addDays(analysisLastTime.getLast(), 1);
        } else {
            ElasticSearchRest.SearchResultClz<AcquireQuotationGoods> data = rest.searchAsClz(BiWarehouseQuotationGoodsName, ElasticSearchRest.Page.of(1, 1), null, Collections.singleton(SortBuilders.fieldSort("time").order(SortOrder.ASC)), null, AcquireQuotationGoods.class);
            if (Objects.isNull(data) || CollectionUtils.isEmpty(data.getData())) {
                throw new BiException("analysisLastTime init error");
            }
            start = Date.from(data.getData().get(0).getTime().atZone(ZoneId.systemDefault()).toInstant());
        }
        Date[] rangeDay = DateUtils.rangeDay(start);
        cache.setLast(start);
        OrganizationQuery query = OrganizationQuery.builder().end(rangeDay[1]).architectureType(1).del(false).build();
        Long count = rest.searchCount(BiWarehouseOrganizationName, BoolQueryBuilders.organization(query));
        log.debug("查询日期 -> {} count -> {}", DateUtils.dateToStr(rangeDay[1], DateUtils.FORMAT_PATTERN_DATE), count);

        //初始化缓存, 不能并行任务
        if (Objects.isNull(cache.getCache())) {
            cache.setCache(new HashMap<>());
        }

        ElasticSearchClzItemReader<AcquireOrganization> reader = new ElasticSearchClzItemReader<>(client, mapper, BiWarehouseOrganizationName,
                count.intValue(), BoolQueryBuilders.organization(query), Collections.singleton(SortBuilders.fieldSort("id").order(SortOrder.ASC)), AcquireOrganization.class);
        reader.setName(ReaderName1);
        return reader;
    }

    @Bean(ProcessName1)
    public ItemProcessor<AcquireOrganization, VendorStatistics> processor1() {
        return acquireOrganization -> {
            Date last = cache.getLast();
            LocalDateTime time = LocalDateTime.ofInstant(last.toInstant(), ZoneId.systemDefault());
            String id = EncryptUtils.MD5_32(DateUtils.dateToStr(last, DateUtils.FORMAT_PATTERN_DATE) + acquireOrganization.getId());
            return VendorStatistics.builder()
                    .id(id)
                    .vendorId(acquireOrganization.getId())
                    .vendorName(acquireOrganization.getCompanyName())
                    .time(time.toLocalDate())
                    .brandNum(0)
                    .goodsNum(0)
                    .missedOrder(0)
                    .orderNum(0)
                    .orderTotal(0.0)
                    .participateNum(0)
                    .purchaseGoodsNum(0)
                    .quotationNum(0)
                    .picketedNum(0)
                    .unitTimeQuotationNum(0)
                    .responsible("")
                    .build();
        };
    }

    @Bean(WriterName1)
    public ItemWriter<VendorStatistics> writer1(){
        return list -> list.forEach(e -> cache.getCache().put(e.getId(), e));
    }

    @Bean(ReaderName2)
    @StepScope
    public ItemReader<VendorStatistics> reader2() {
        return EngineUtils.createCacheReader(cache.getCache());
    }

    @Bean(ProcessName2)
    public ItemProcessor<VendorStatistics, VendorStatistics> processor2() {
        return vendorStatistics -> {
            Long vendorId = vendorStatistics.getVendorId();
            Date[] rangeDay = DateUtils.rangeDay(cache.getLast());
            GoodsQuery query = GoodsQuery.builder().del(false).start(rangeDay[0]).end(rangeDay[1]).vendorId(vendorId).build();
            BoolQueryBuilder queryBuilder = BoolQueryBuilders.goods(query);
            long count = rest.searchCount(BiWarehouseGoodsName, queryBuilder);
            vendorStatistics.setGoodsNum((int) count);

            TimeValue scroll = TimeValue.timeValueMinutes(3L);
            SearchResponse response = rest.search(BiWarehouseGoodsName, ElasticSearchRest.Page.of(1, 100),
                    queryBuilder, Collections.singletonList(SortBuilders.fieldSort("id").order(SortOrder.ASC)), null, scroll);
            List<AcquireGoods> acquireGoods = EngineUtils.scrollSearch(response, rest, mapper, scroll, AcquireGoods.class);

            if (!CollectionUtils.isEmpty(acquireGoods)) {
                List<Long> profileIds = acquireGoods.stream().map(AcquireGoods::getProfileId).collect(Collectors.toList());
                GoodsProfileQuery profileQuery = GoodsProfileQuery.builder().ids(profileIds).build();
                BoolQueryBuilder boolQueryBuilder = BoolQueryBuilders.goodsProfile(profileQuery);
                AggregationBuilder brandNum = AggregationBuilders.cardinality("brand").field("brand").precisionThreshold(10000);
                SearchResponse search = rest.search(BiWarehouseGoodsProfileName, ElasticSearchRest.Page.of(1, 0), boolQueryBuilder, null, Collections.singletonList(brandNum), null);
                Map<String, Aggregation> stringAggregationMap = search.getAggregations().asMap();
                Aggregation brand = stringAggregationMap.get("brand");
//            log.debug("goods aggregation brand {}", brand);
                if (Objects.nonNull(brand)) {
                    vendorStatistics.setBrandNum(((int) (((Cardinality) brand).getValue())));
                }
            }
            return vendorStatistics;
        };
    }

    @Bean(ProcessName3)
    public ItemProcessor<VendorStatistics, VendorStatistics> processor3() {
        return vendorStatistics -> {
            Long vendorId = vendorStatistics.getVendorId();
            Date[] rangeDay = DateUtils.rangeDay(cache.getLast());

            QuotationGoodsQuery participateNumQuery = QuotationGoodsQuery.builder().vendorId(vendorId).del(false).start(rangeDay[0]).end(rangeDay[1]).build();
            AggregationBuilder participateNumAgg = AggregationBuilders.filter(PARTICIPATE_NUM, BoolQueryBuilders.quotationGoods(participateNumQuery));
            participateNumAgg.subAggregation(AggregationBuilders.count(ID_NUM).field("id"));

            QuotationGoodsQuery quotationNumQuery = participateNumQuery.clone();
            quotationNumQuery.setStatuses(Arrays.asList(3, 4, 5, 6, 7));
            AggregationBuilder quotationNumAgg = AggregationBuilders.filter(QUOTATION_NUM, BoolQueryBuilders.quotationGoods(quotationNumQuery));
            quotationNumAgg.subAggregation(AggregationBuilders.count(ID_NUM).field("id"));

            QuotationGoodsQuery picketedNumQuery = participateNumQuery.clone();
            picketedNumQuery.setStatuses(Collections.singletonList(5));
            AggregationBuilder picketedNumAgg = AggregationBuilders.filter(PICKETED_NUM, BoolQueryBuilders.quotationGoods(picketedNumQuery));
            picketedNumAgg.subAggregation(AggregationBuilders.count(ID_NUM).field("id"));

            QuotationGoodsQuery unitTimeQuotationNumQuery = participateNumQuery.clone();
            unitTimeQuotationNumQuery.setStatuses(Arrays.asList(3, 4)).setLastLtTimeAfterThirtySixHour(true);
            AggregationBuilder unitTimeQuotationNumAgg = AggregationBuilders.filter(UNIT_TIME_QUOTATION_NUM, BoolQueryBuilders.quotationGoods(unitTimeQuotationNumQuery));
            unitTimeQuotationNumAgg.subAggregation(AggregationBuilders.count(ID_NUM).field("id"));

            List<AggregationBuilder> aggregations = Arrays.asList(participateNumAgg, quotationNumAgg, picketedNumAgg, unitTimeQuotationNumAgg);
            SearchResponse response = rest.search(BiWarehouseQuotationGoodsName, ElasticSearchRest.Page.of(1, 0), null, null, aggregations, null);

            Map<String, Aggregation> stringAggregationMap = response.getAggregations().asMap();
            HashMap<String, Long> aggResult = stringAggregationMap.entrySet().stream().collect(HashMap::new, (m, v) ->
                            m.put(v.getKey(), ((ValueCount) (((ParsedFilter) v.getValue()).getAggregations().asMap().get(ID_NUM))).getValue()),
                    HashMap::putAll);
            vendorStatistics.setQuotationNum(aggResult.getOrDefault(QUOTATION_NUM, 0L).intValue())
                    .setParticipateNum(aggResult.getOrDefault(PARTICIPATE_NUM, 0L).intValue())
                    .setPicketedNum(aggResult.getOrDefault(PICKETED_NUM, 0L).intValue())
                    .setUnitTimeQuotationNum(aggResult.getOrDefault(UNIT_TIME_QUOTATION_NUM, 0L).intValue());
            vendorStatistics.setPurchaseGoodsNum(vendorStatistics.getPicketedNum());
            return vendorStatistics;
        };
    }

    @Bean(ProcessName4)
    public ItemProcessor<VendorStatistics, VendorStatistics> processor4() {
        return vendorStatistics -> {
            Long vendorId = vendorStatistics.getVendorId();
            Date[] rangeDay = DateUtils.rangeDay(cache.getLast());
            VendorOrderQuery query = VendorOrderQuery.builder().start(rangeDay[0]).end(rangeDay[1]).vendorId(vendorId).del(false).build();
            BoolQueryBuilder boolQueryBuilder = BoolQueryBuilders.vendorOrder(query);
            AggregationBuilder id = AggregationBuilders.count(ID_NUM).field("id");
            AggregationBuilder amount = AggregationBuilders.sum("amount").field("amount");

            SearchResponse search = rest.search(BiWarehouseVendorOrderName, null, boolQueryBuilder, null, Arrays.asList(id, amount), null);
            Map<String, Aggregation> stringAggregationMap = search.getAggregations().asMap();
            Aggregation orderNum = stringAggregationMap.get(ID_NUM);
            Aggregation orderTotal = stringAggregationMap.get("amount");
            if (Objects.nonNull(orderNum)) {
                vendorStatistics.setOrderNum((int) ((ValueCount) orderNum).getValue());
            }
            if (Objects.nonNull(orderTotal)) {
                vendorStatistics.setOrderTotal(((Sum) orderTotal).getValue());
            }
            return vendorStatistics;
        };
    }

    @Bean(ProcessName5)
    public ItemProcessor<VendorStatistics, VendorStatistics> processor5() {
        return vendorStatistics -> {
            Long vendorId = vendorStatistics.getVendorId();
            Date[] rangeDay = DateUtils.rangeDay(cache.getLast());
            VendorOrderQuery query = VendorOrderQuery.builder().start(rangeDay[0]).end(rangeDay[1]).vendorId(vendorId).lastLtTimeAfterTwelveHour(true).del(false).build();
            BoolQueryBuilder boolQueryBuilder = BoolQueryBuilders.vendorOrder(query);
            AggregationBuilder id = AggregationBuilders.count(ID_NUM).field("id");

            SearchResponse search = rest.search(BiWarehouseVendorOrderName, ElasticSearchRest.Page.of(1, 0), boolQueryBuilder, null, Collections.singletonList(id), null);
            Map<String, Aggregation> stringAggregationMap = search.getAggregations().asMap();
            Aggregation orderNum = stringAggregationMap.get(ID_NUM);
            if (Objects.nonNull(orderNum)) {
                vendorStatistics.setMissedOrder((int) ((ValueCount) orderNum).getValue());
            }
            return vendorStatistics;
        };
    }

//    @Bean(ProcessName6)
//    public ItemProcessor<VendorStatistics, VendorStatistics> processor6() {
//        return vendorStatistics -> {
//            Long vendorId = vendorStatistics.getVendorId();
//            Date[] rangeDay = DateUtils.rangeDay(cache.getLast());
//            PurchaseGoodsQuery query = PurchaseGoodsQuery.builder().start(rangeDay[0]).end(rangeDay[1]).vendorId(vendorId).del(false).build();
//            BoolQueryBuilder boolQueryBuilder = BoolQueryBuilders.purchaseGoods(query);
//            AggregationBuilder id = AggregationBuilders.count(ID_NUM).field("id");
//
//            SearchResponse search = rest.search(BiWarehousePurchaseGoodsName, ElasticSearchRest.Page.of(1, 0), boolQueryBuilder, null, Collections.singletonList(id), null);
//            Map<String, Aggregation> stringAggregationMap = search.getAggregations().asMap();
//            Aggregation goodsNum = stringAggregationMap.get(ID_NUM);
//            if (Objects.nonNull(goodsNum)) {
//                vendorStatistics.setPurchaseGoodsNum((int) ((ValueCount) goodsNum).getValue());
//            }
//            return vendorStatistics;
//        };
//    }

    @Bean(ProcessName7)
    public ItemProcessor<VendorStatistics, VendorStatistics> processor7() {
        return vendorStatistics -> {
            BoolQueryBuilder query = BoolQueryBuilders.platformGrantUser(vendorStatistics.getVendorId());
            long count = rest.searchCount(BiWarehousePlatformGrantUserName, query);
            ElasticSearchRest.SearchResultClz<AcquirePlatformGrantUser> result = rest.searchAsClz(BiWarehousePlatformGrantUserName, ElasticSearchRest.Page.of(1, (int) count), query, null, null, AcquirePlatformGrantUser.class);
            if (Objects.isNull(result) || CollectionUtils.isEmpty(result.getData())) {
                return vendorStatistics;
            }
            Set<Long> memberIds = result.getData().stream().filter(Objects::nonNull).map(AcquirePlatformGrantUser::getMemberId).collect(Collectors.toSet());

            BoolQueryBuilder queryBuilder = BoolQueryBuilders.member(MemberQuery.builder().ids(memberIds).build());
            ElasticSearchRest.SearchResultClz<AcquireMember> memberResult = rest.searchAsClz(BiWarehouseMemberName, ElasticSearchRest.Page.of(1, memberIds.size()), queryBuilder, null, null, AcquireMember.class);
            if (Objects.isNull(memberResult) || CollectionUtils.isEmpty(memberResult.getData())) {
                return vendorStatistics;
            }
            Set<String> memberNames = memberResult.getData().stream().filter(Objects::nonNull).map(AcquireMember::getFullname).collect(Collectors.toSet());
            return vendorStatistics.setResponsible(String.join(",", memberNames));
        };
    }




    @Bean(WriterName2)
    public ElasticSearchClzItemWriter<VendorStatistics> writer2() {

        return new ElasticSearchClzItemWriter<>(client, mapper, BiDataMartName);
    }

    @Bean(JobName)
    public Job job(@Qualifier(StepName1) Step step1, @Qualifier(StepName2) Step step2,@Qualifier(StepName3) Step step3,
                   @Qualifier(StepName3) Step step4, @Qualifier(StepName4) Step step5, @Qualifier(StepName7) Step step7) {

        return jobBuilderFactory.get(JobName)
                .incrementer(new RunIdIncrementer())
                .start(step1)
                .next(step2)
                .next(step3)
                .next(step4)
                .next(step5)
//                .next(step6)
                .next(step7)
                .listener(jobExecutionListener)
                .build();
    }

    @Bean(StepName1)
    public Step step1() {

        return stepBuilderFactory.get(StepName1)
                .<AcquireOrganization, VendorStatistics>chunk(MAX_BATCH_NUM)
                .reader(reader1())
                .processor(processor1())
                .writer(writer1())
                .build();
    }

    @Bean(StepName2)
    public Step step2(@Qualifier(ReaderName2)ItemReader<? extends VendorStatistics> reader2) {

        return stepBuilderFactory.get(StepName2)
                .<VendorStatistics, VendorStatistics>chunk(MAX_BATCH_NUM)
                .reader(reader2)
                .processor(processor2())
                .writer(writer1())
                .build();
    }

    @Bean(StepName3)
    public Step step3(@Qualifier(ReaderName2)ItemReader<? extends VendorStatistics> reader2) {

        return stepBuilderFactory.get(StepName3)
                .<VendorStatistics, VendorStatistics>chunk(MAX_BATCH_NUM)
                .reader(reader2)
                .processor(processor3())
                .writer(writer1())
                .build();
    }

    @Bean(StepName4)
    public Step step4(@Qualifier(ReaderName2)ItemReader<? extends VendorStatistics> reader2) {

        return stepBuilderFactory.get(StepName4)
                .<VendorStatistics, VendorStatistics>chunk(MAX_BATCH_NUM)
                .reader(reader2)
                .processor(processor4())
                .writer(writer1())
                .build();
    }

    @Bean(StepName5)
    public Step step5(@Qualifier(ReaderName2)ItemReader<? extends VendorStatistics> reader2) {

        return stepBuilderFactory.get(StepName5)
                .<VendorStatistics, VendorStatistics>chunk(MAX_BATCH_NUM)
                .reader(reader2)
                .processor(processor5())
                .writer(writer1())
                .build();
    }

//    @Bean(StepName6)
//    public Step step6(@Qualifier(ReaderName2)ItemReader<? extends VendorStatistics> reader2) {
//
//        return stepBuilderFactory.get(StepName6)
//                .<VendorStatistics, VendorStatistics>chunk(MAX_BATCH_NUM)
//                .reader(reader2)
//                .processor(processor6())
//                .writer(writer1())
//                .build();
//    }

    @Bean(StepName7)
    public Step step7(@Qualifier(ReaderName2)ItemReader<? extends VendorStatistics> reader2) {

        return stepBuilderFactory.get(StepName7)
                .<VendorStatistics, VendorStatistics>chunk(MAX_BATCH_NUM)
                .reader(reader2)
                .processor(processor7())
                .writer(writer2())
                .build();
    }
}
